#pragma once
#include <ev.h>
#include <amqpcpp.h>
#include <amqpcpp/libev.h>
#include <openssl/ssl.h>
#include <openssl/opensslv.h>

#include "log.hpp"

class RabbitMQClient {
public:
    using ptr = std::shared_ptr<RabbitMQClient>;
    using MessageCallback = std::function<void(const char*, size_t)>;
public:
    // host为服务器的IP地址加端口 "127.0.0.1:5672"
    RabbitMQClient(const std::string &user,
                   const std::string &password,
                   const std::string &host) {
        _loop = EV_DEFAULT;
        _handler = std::make_unique<AMQP::LibEvHandler>(_loop);
        AMQP::Address address("amqp://" + user + ":" + password + "@" + host + "/");
        _connection = std::make_unique<AMQP::TcpConnection>(_handler.get(), address);
        _channel = std::make_unique<AMQP::TcpChannel>(_connection.get());

        _loop_thread = std::thread([this]() {
            ev_run(_loop, 0);
        });
    }

    //void (*cb)(struct ev_loop *loop, struct ev_async *w, int revents);
    ~RabbitMQClient() {
        ev_async_init(&_watcher, watcher_cb);
        ev_async_start(_loop, &_watcher);
        ev_async_send(_loop, &_watcher);
        if (_loop_thread.joinable())
            _loop_thread.join();
        _loop = nullptr;
    }




    void declareComponents(const std::string &exchange,
                           const std::string &queue,
                           const std::string &routingkey = "routing_key",
                           AMQP::ExchangeType type = AMQP::ExchangeType::direct) {
        // 声明交换机
        _channel->declareExchange(exchange, type)
            .onError([](const char *message)->void {
                LOG_ERROR("声明交换机失败 ",  message);
                exit(-1); 
            })
            .onSuccess([]()->void  { 
                LOG_INFO("交换机声明成功"); 
            });


        // 声明队列
        _channel->declareQueue(queue)
            .onError([](const char *message)->void {
                LOG_ERROR("声明队列失败 ", message);;
                exit(-1); 
            })
            .onSuccess([]()->void { 
                LOG_INFO("队列声明成功 "); 
            });


        // 绑定交换机和队列
        _channel->bindQueue(exchange, queue, routingkey)
            .onError([](const char *message)->void {
                LOG_ERROR("绑定交换机和队列失败 ", message);
                exit(-1); 
            })
            .onSuccess([exchange, queue, routingkey]()->void { 
                LOG_INFO("绑定成功 {}-{}-{}", exchange, queue, routingkey);            
            });
    }


    bool publish(const std::string &exchange, 
                 const std::string &message, 
                 const std::string &routingKey = "routing_key") {
        bool success = _channel->publish(exchange, routingKey, message);
        if (!success) {
            LOG_ERROR("发布失败");
            return false;
        }
        LOG_INFO("向交换机 {}-{} 发布 {} 成功", exchange, routingKey, message);
        return true;
    }


    void consume(const std::string& queue, const MessageCallback& cb) {
        _channel->consume(queue).
        onReceived([this, cb](const AMQP::Message &message, uint64_t deliveryTag, bool redelivered)->void {
            cb(message.body(), message.bodySize());
            bool success = _channel->ack(deliveryTag);
            if(!success) {
                LOG_ERROR("ack消息失败");
                return;
            }
        }).
        onError([&queue](const char *message)->void {
            LOG_ERROR("订阅 {} 队列 {} 消息失败 ", queue, message);
        });
    }
private:
    //void (*cb)(struct ev_loop *loop, struct ev_async *w, int revents);
    static void watcher_cb(struct ev_loop *loop, struct ev_async *w, int revents) {
        ev_break(loop, EVBREAK_ALL);
    }


private:
    struct ev_async _watcher;
    struct ev_loop *_loop;
    std::unique_ptr<AMQP::LibEvHandler> _handler;
    std::unique_ptr<AMQP::TcpConnection> _connection;
    std::unique_ptr<AMQP::TcpChannel> _channel;
    std::thread _loop_thread;
};